package com.sy.rabbitmq.dlx;

import com.rabbitmq.client.*;
import com.sy.rabbitmq.TestProperties;

import java.io.IOException;
import java.util.HashMap;
import java.util.concurrent.TimeoutException;

/**
 * dlx-死信队列-消费端
 * 消息超时和队列超时和nack时设置不重回队列都会进入死信队列中
 */
public class Comsumer {
    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setUsername(TestProperties.getUsername());
        factory.setPassword(TestProperties.getPassword());
        factory.setHost(TestProperties.getIp());
        factory.setPort(TestProperties.getPort());
        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();
        //todo 4.声明一个普通的队列和路由
        String exchangeName = "test_dlx_exchange";
        String routeKey = "dlx.#";
        String queueName = "test_dlx_queue";

        //声明交换机
        channel.exchangeDeclare(exchangeName,"topic",true,false,null);

        //todo 配合死信队列配置到队列上 因为是队列绑定交换机 重新投发到死信中
        HashMap<String, Object> arguments = new HashMap<String, Object>();
        arguments.put("x-dead-letter-exchange","dlx.exchange");
        //声明队列
        channel.queueDeclare(queueName,true,false,false,arguments);
        //将交换机和队列以及路由key(规则)绑定一起
        channel.queueBind(queueName,exchangeName,routeKey);

        //todo 进行死信队列声明:
         channel.exchangeDeclare("dlx.exchange","topic",true,false,null);
        channel.queueDeclare("dlx.queue",true,false,false,null);
        //todo 必须要设置路由key为通配
        channel.queueBind("dlx.queue","dlx.exchange","#");

        //6.设置qos 限流
        //prefetchSize 消息大小限制
        //prefetchCount 消息数量 在ack没有回复前多大承受消息数量
        //global 设置限流是应用到整个channel还是当前consume
        channel.basicQos(0,1,false);
        //7.限流方式必须自己手动ack不能自动ack
        channel.basicConsume(queueName,false,new MyConsumer(channel));
    }

    private static class MyConsumer extends DefaultConsumer {
        private Channel channel;
        /**
         * Constructs a new instance and records its association to the passed-in channel.
         *
         * @param channel the channel to which this consumer is attached
         */
        public MyConsumer(Channel channel) {
            super(channel);
            this.channel = channel;
        }


        /**
         * 具体消费代码编写
         * @param consumerTag
         * @param envelope
         * @param properties
         * @param body
         * @throws IOException
         */
        @Override
        public void handleDelivery(String consumerTag, Envelope envelope, AMQP.BasicProperties properties, byte[] body) throws IOException {
            System.out.println("-----------------新消费方式-----------------");
//            System.out.println("consumerTag"+consumerTag);
//            System.out.println("envelope"+envelope);
//            System.out.println("AMQP.BasicProperties"+properties);
            System.out.println("body"+new String(body));
            try {
                Thread.sleep(1000);
            } catch (InterruptedException e) {
                System.out.println(e.getMessage());
            }
            int num = ((int)properties.getHeaders().get("num"));
            if (num%2==0){
                System.out.println("ack:"+num);
                //回复ack
                //basicAck(deliveryTag,multiple)
                //deliveryTag 消息标识
                //multiple 是否批量回复
                this.channel.basicAck(envelope.getDeliveryTag(),false);
            }else{
                System.out.println("nack:"+num);
                properties.getHeaders().put("num",++num);
                //回复nack
                //basicAck(deliveryTag,multiple,requeue)
                //deliveryTag: 消息标识
                //multiple: 是否批量回复
                //todo requeue:表示是否重回队列-如果设置为false那么就会投递到死信队列中
                this.channel.basicNack(envelope.getDeliveryTag(),false,false);
            }
            System.out.println("------------------------------------");
        }
    }
}
